package com.belf.console.service.impl;

import com.alibaba.fastjson2.JSONObject;
import com.belf.common.config.MigrateConfig;
import com.belf.common.core.domain.AjaxResult;
import com.belf.common.core.domain.entity.SysUser;
import com.belf.common.utils.DateUtils;
import com.belf.common.utils.SecurityUtils;
import com.belf.console.agent.migrate.job.CheckJob;
import com.belf.console.domain.DataCheckTask;
import com.belf.console.domain.DataSourceInfo;
import com.belf.console.domain.DataTaskLog;
import com.belf.console.mapper.DataCheckTaskMapper;
import com.belf.console.service.IBelfTaskJobService;
import com.belf.console.service.IDataCheckTaskService;
import com.belf.console.service.IDataSourceInfoService;
import com.belf.console.service.IDataTaskLogService;
import com.belf.quartz.service.ISysJobService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.List;

import static belf.migrate.engine.admin.taskconf.JobCode.UNKNOW_ERROR;
import static com.belf.common.constant.BelfCodeConstants.*;
import static com.belf.common.core.domain.AjaxResult.DATA_TAG;

/**
 * 校验任务Service业务层处理
 * 
 * @author belf
 * @date 2024-07-16
 */
@Service
public class DataCheckTaskServiceImpl implements IDataCheckTaskService {
    private static final Logger LOGGER = LoggerFactory.getLogger(DataCheckTaskServiceImpl.class.getName());
    @Autowired
    private DataCheckTaskMapper dataCheckTaskMapper;
    @Autowired
    private MigrateConfig migrateConfig;
    @Autowired
    private IDataTaskLogService dataTaskLogService;
    @Autowired
    private IDataSourceInfoService dataSourceInfoService;
    @Autowired
    private ISysJobService sysJobService;
    @Autowired
    private IBelfTaskJobService belfTaskJobService;

    /**
     * 查询校验任务
     * 
     * @param id 校验任务主键
     * @return 校验任务
     */
    @Override
    public DataCheckTask selectDataCheckTaskById(Long id)
    {
        return dataCheckTaskMapper.selectDataCheckTaskById(id);
    }

    /**
     * 查询校验任务列表
     * 
     * @param dataCheckTask 校验任务
     * @return 校验任务
     */
    @Override
    public List<DataCheckTask> selectDataCheckTaskList(DataCheckTask dataCheckTask)
    {
        return dataCheckTaskMapper.selectDataCheckTaskList(dataCheckTask);
    }

    @Override
    public long selectDataCount(DataCheckTask dataCheckTask) {

        return dataCheckTaskMapper.selectDataCount(dataCheckTask);
    }

    /**
     * 新增校验任务
     * 
     * @param task 校验任务
     * @return 结果
     */
    @Override
    public int insertDataCheckTask(DataCheckTask task)
    {
        task.setCreateBy(SecurityUtils.getLoginUser().getUsername());
        task.setCreateTime(DateUtils.getNowDate());

        task.setUpdateBy(SecurityUtils.getLoginUser().getUsername());
        task.setUpdateTime(DateUtils.getNowDate());
        task.setTaskStatus(BELF_TASK_JOB_STATUS_INIT);
        int i = dataCheckTaskMapper.insertDataCheckTask(task);
        if (task.getStrategy() != null && task.getStrategy()==BELF_TASK_JOB_CREATE_RUN) {
            executeTaskById(task.getId());
        }
        belfTaskJobService.insertJob(task.getMonitorConf(),task.getId(),task.getTaskName(),task.getTaskType(),task.getMonitorStatus());
        return i;
    }



    /**
     * 修改校验任务
     * 
     * @param task 校验任务
     * @return 结果
     */
    @Override
    public int updateDataCheckTask(DataCheckTask task)
    {
        task.setUpdateTime(DateUtils.getNowDate());
        int i = dataCheckTaskMapper.updateDataCheckTask(task);
        task = selectDataCheckTaskById(task.getId());
        belfTaskJobService.deleteTaskJob(task.getId(),task.getTaskType());
        // 删除旧的定时任务，添加新的定时任务
        belfTaskJobService.insertJob(task.getMonitorConf(),task.getId(),task.getTaskName(),task.getTaskType(),task.getMonitorStatus());
        return i;
    }

    public int updateData(DataCheckTask task)
    {
        task.setUpdateTime(DateUtils.getNowDate());
        return dataCheckTaskMapper.updateDataCheckTask(task);
    }

    /**
     * 批量删除校验任务
     * 
     * @param ids 需要删除的校验任务主键
     * @return 结果
     */
    @Override
    public int deleteDataCheckTaskByIds(Long[] ids)
    {
        if (ids != null) {
            for (Long taskId : ids) {
                deleteDataCheckTaskById(taskId);
            }
        }
        return 1;
//        return dataCheckTaskMapper.deleteDataCheckTaskByIds(ids);
    }

    /**
     * 删除校验任务信息
     * 
     * @param id 校验任务主键
     * @return 结果
     */
    @Override
    public int deleteDataCheckTaskById(Long id)
    {
        try {
            DataCheckTask appraiseTasks = selectDataCheckTaskById(id);
            belfTaskJobService.deleteTaskJob(id,appraiseTasks.getTaskType());
            return dataCheckTaskMapper.deleteDataCheckTaskById(id);
        } catch (Exception e) {
            LOGGER.error("删除定时任务失败", e);
        }
//        return dataCheckTaskMapper.deleteDataCheckTaskById(id);
        return 0;
    }

    @Override
    public AjaxResult executeTaskById(Long id) {

        DataCheckTask task = selectDataCheckTaskById(id);
        DataTaskLog jobLog = createTaskLog(task);
        // 提交评估任务
        try {
            pushJob(task,jobLog);
            return AjaxResult.success();
        } catch (Exception e) {
            LOGGER.error("数据源信息 连接测试失败",e);
        }
        return AjaxResult.error();
    }


    @Override
    public AjaxResult schedulerTask(Long[] idArray) {
        return AjaxResult.success();
    }

    @Override
    public HashMap<String, HashMap<String, Integer>> selectParseCount(SysUser user) {
        DataCheckTask task = new DataCheckTask();
        if (!user.isAdmin()){
            task.setCreateBy(user.getUserName());
        }
        HashMap<String,  HashMap<String, Integer>> resultMap = new HashMap<>();
        // 根据任务类型统计
        resultMap.put("dataCheckJobTypeCount", dataCheckTaskMapper.queryJobTypeCount(task));
        // 根据任务状态统计
        resultMap.put("dataCheckTaskStatusCount",dataCheckTaskMapper.queryTaskStatusCount(task));
        return resultMap;
    }

    @Override
    public DataCheckTask selectDataCheckTaskByIdAndTaskType(DataCheckTask dataCheckTask) {

        return dataCheckTaskMapper.selectDataCheckTaskByIdAndTaskType(dataCheckTask);
    }

    private void pushJob(DataCheckTask task,DataTaskLog jobLogParam){
        new Thread(()->{
            LOGGER.info("pushJob");
            DataTaskLog jobLog = jobLogParam;
            try {
                AjaxResult ajaxResult = CheckJob.pushJob(migrateConfig.getPushJob(), task,  jobLog);
                LOGGER.info("任务返回信息：{}", ajaxResult);
                if (ajaxResult.isSuccess()) {
                    JSONObject resultData = JSONObject.from(ajaxResult.get(DATA_TAG));
                    Long taskStatus = resultData.getLong(AjaxResult.CODE_TAG);

                    jobLog = dataTaskLogService.selectDataTaskLogById(jobLog.getId());

                    jobLog.setTaskResult(ajaxResult.get(DATA_TAG).toString());
                    jobLog.setTaskStatus(taskStatus);
                    task.setTaskStatus(taskStatus);
                }else {
                    task.setTaskStatus((Long) ajaxResult.get(AjaxResult.CODE_TAG));
                    jobLog.setTaskStatus((Long) ajaxResult.get(AjaxResult.CODE_TAG));
                }
                if (ajaxResult.get("conf") != null) {
                    jobLog.setTaskConfig(ajaxResult.get("conf").toString());
                }
            }catch (Exception e){
                LOGGER.error("任务提交失败",e);
                //  判断返回结果状态 更新数据库
                task.setTaskStatus(UNKNOW_ERROR);
                jobLog.setTaskStatus( UNKNOW_ERROR);
            }
            jobLog.setEndTime(DateUtils.getNowDate());
            jobLog.setTimeConsuming(DateUtils.timeDistance(jobLog.getEndTime(),jobLog.getStartTime()));

            updateData(task);
            dataTaskLogService.saveDataTaskLogResult(jobLog);
        }).start();
    }

    public DataTaskLog createTaskLog(DataCheckTask task){
        DataTaskLog jobLog = new DataTaskLog();
        jobLog.setTaskType(task.getTaskType());
        jobLog.setTaskId(task.getId());
        jobLog.setCreateBy(task.getUpdateBy());
        jobLog.setCreateTime(DateUtils.getNowDate());
        jobLog.setJobType(task.getJobType());
        jobLog.setTaskStatus(BELF_TASK_JOB_STATUS_RUNING);
        jobLog.setTaskName(task.getTaskName());
        dataTaskLogService.insertDataTaskLog(jobLog);

        String sourceId = task.getSourceId();
        String sinkId = task.getSinkId();
        String[] ids={sourceId,sinkId};
        List<DataSourceInfo> dataSourceList=dataSourceInfoService.selectDataSourceInfoByIds(ids);
        dataSourceList.forEach(dataSourceInfo -> {
            if (dataSourceInfo.getId().equals(sourceId)){
//                dataSourceInfoService.setDataSourceDatabase(dataSourceInfo, task.getSourceDatabase());
                dataSourceInfo.setDatabaseName(task.getSourceDatabase());
                dataSourceInfo.setSchemaName(task.getSourceDataSchema());
            }
            if (dataSourceInfo.getId().equals(sinkId)){
//                dataSourceInfoService.setDataSourceDatabase(dataSourceInfo, task.getSinkDatabase());
                dataSourceInfo.setDatabaseName(task.getSinkDatabase());
                dataSourceInfo.setSchemaName(task.getSinkDataSchema());
            }
        });
        task.setTaskStatus(BELF_TASK_JOB_STATUS_RUNING);
        task.setDataSourceInfoList(dataSourceList);
        updateData(task);
        return jobLog;
    }
}
